package cn.mikeal.clicksession;

/*
 * 获取点击流的表
 */
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.Locale;
import java.util.UUID;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import cn.mikeal.bean.WebLogBean;

public class ClickSessionStream {

	static class ClickSessionStreamMapper extends Mapper<LongWritable, Text, Text, WebLogBean> {
		private Text mk = new Text();

		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, WebLogBean>.Context context)
				throws IOException, InterruptedException {
			String[] datas = value.toString().split("\001");
			if (datas.length == 9) {
				WebLogBean bean = new WebLogBean();
				bean.setIs_valid(datas[0].equals("true") ? true : false);
				bean.setRemote_addr(datas[1]);
				bean.setRemote_user(datas[2]);
				bean.setTime_local(datas[3]);
				bean.setRequest(datas[4]);
				bean.setStatus(datas[5]);
				bean.setBody_bytes_sent(datas[6]);
				bean.setHttp_referer(datas[7]);
				bean.setHttp_user_agent(datas[8]);
				// 数据需要过滤的
				if (bean.isIs_valid()) {
					mk.set(bean.getRemote_addr());
					context.write(mk, bean);
				}
			}

		}

	}

	static class ClickSessionStreamReducer extends Reducer<Text, WebLogBean, Text, NullWritable> {
		private Text rk = new Text();

		@Override
		protected void reduce(Text key, Iterable<WebLogBean> values,
				Reducer<Text, WebLogBean, Text, NullWritable>.Context context)
				throws IOException, InterruptedException {
			String sessionid = UUID.randomUUID().toString().replace("-", "");
			ArrayList<WebLogBean> lists = new ArrayList<>();
			// 将数据存储在集合中
			for (WebLogBean v : values) {
				// 对象重用： 注意：key随着value变化而变化的，且key只有一个内存对象。 且value只有一个内存对象
				WebLogBean bean = new WebLogBean();
				try {
					BeanUtils.copyProperties(bean, v);
				} catch (IllegalAccessException e) {
					e.printStackTrace();
				} catch (InvocationTargetException e) {
					e.printStackTrace();
				}
				// bean = v;
				lists.add(bean);
			}
			// 将集合按照时间排序
			Collections.sort(lists, new Comparator<WebLogBean>() {

				@Override
				public int compare(WebLogBean bean1, WebLogBean bean2) {
					Date d1 = null;
					Date d2 = null;
					try {
						d1 = toDate(bean1.getTime_local());
						d2 = toDate(bean2.getTime_local());
					} catch (ParseException e) {
						e.printStackTrace();
					}
					if (d1 == null) {
						return -1;
					} else if (d2 == null) {
						return 1;
					} else {
						return d1.compareTo(d2);
					}
				}
			});
			int step = 1;
			int staytimes = 60;
			// 第一种情况，session下只有一次页面访问,设置页面停留时间为默认的60
			if (lists.size() == 1) {
				rk.set(sessionid + "\001" + lists.get(0).getRemote_addr() + "\001" + lists.get(0).getTime_local()
						+ "\001" + lists.get(0).getBody_bytes_sent() + "\001" + lists.get(0).getHttp_referer() + "\001"
						+ (60) + "\001" + (1));
				context.write(rk, NullWritable.get());
			} else {
				// 第二种情况，session下多次页面访问，遍历集合
				for (int i = 0; i < lists.size(); i++) {
					if (i == 0) { // 同一个ip的第一个数据
						rk.set(sessionid + "\001" + lists.get(0).getRemote_addr() + "\001"
								+ lists.get(0).getTime_local() + "\001" + lists.get(0).getBody_bytes_sent() + "\001"
								+ lists.get(0).getHttp_referer() + "\001" + (60) + "\001" + (1));
						context.write(rk, NullWritable.get());
					}
					Long difftimes = 0L;
					try {
						difftimes = diffDate(lists.get(i - 1).getTime_local(), lists.get(i).getTime_local());
					} catch (ParseException e) {
						e.printStackTrace();
					}
					if (difftimes < 30 * 60 * 1000) { // 同一个会话
						staytimes = (int) (difftimes / 1000);
						rk.set(sessionid + "\001" + lists.get(i).getRemote_addr() + "\001"
								+ lists.get(i).getTime_local() + "\001" + lists.get(i).getBody_bytes_sent() + "\001"
								+ lists.get(i).getHttp_referer() + "\001" + staytimes + "\001" + step);
						context.write(rk, NullWritable.get());
						step++;
					} else { // 新的会话 -- 生成新的sessionid
						sessionid = UUID.randomUUID().toString().replace("-", "");
						staytimes = (int) (difftimes / 1000);
						rk.set(sessionid + "\001" + lists.get(i).getRemote_addr() + "\001"
								+ lists.get(i).getTime_local() + "\001" + lists.get(i).getBody_bytes_sent() + "\001"
								+ lists.get(i).getHttp_referer() + "\001" + staytimes + "\001" + step);
						context.write(rk, NullWritable.get());
						step++;
					}

					if (i == lists.size()) { // 同一个ip的最后一条数据
						rk.set(sessionid + "\001" + lists.get(0).getRemote_addr() + "\001"
								+ lists.get(0).getTime_local() + "\001" + lists.get(0).getBody_bytes_sent() + "\001"
								+ lists.get(0).getHttp_referer() + "\001" + (60) + "\001" + (1));
						context.write(rk, NullWritable.get());
					}

				}
			}

		}

	}

	// 计算时间
	public static Date toDate(String times) throws ParseException {
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.US);
		return sdf.parse(times);
	}

	// 计算两个时间差
	public static Long diffDate(String date1, String date2) throws ParseException {
		Date d1 = toDate(date1);
		Date d2 = toDate(date2);
		return d1.getTime() - d2.getTime();
	}

	public static void main(String[] args) {

	}

}
